package com.atguigu.day04;

import com.atguigu.bean.OrderEvent;
import com.atguigu.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Flink05_Project_Order {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.分别获取两张表的数据，并转为JavaBean
        SingleOutputStreamOperator<OrderEvent> orderEventStream = env.readTextFile("input/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new OrderEvent(Long.parseLong(split[0]), split[1], split[2], Long.parseLong(split[3]));
                    }
                });

        SingleOutputStreamOperator<TxEvent> txEventStream = env.readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new TxEvent(split[0], split[1], Long.parseLong(split[2]));
                    }
                });

        //3.连接两条流
        ConnectedStreams<OrderEvent, TxEvent> connect = orderEventStream.connect(txEventStream);

        //4.将相同txId的数据聚合到一块  在多并行度的情况下要做keyBy
        ConnectedStreams<OrderEvent, TxEvent> keyedStream = connect.keyBy("txId", "txId");

        //5.实时对账两个流里面的数据，看是否能够根据txId关联上
        keyedStream.process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {

            //创建存放订单数据的Map集合
            private HashMap<String, OrderEvent> orderMap = new HashMap<>();

            //创建存放交易数据的Map集合
            private HashMap<String, TxEvent> txMap = new HashMap<>();

            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                //1.先判断对方缓存中是否有能关联上的数据
                if (txMap.containsKey(value.getTxId())) {
                    //有：关联成功，结果打印至控制台
                    out.collect("订单：" + value.getOrderId() + "对账成功！！！！！");
                    //并且移除对方缓存中已关联上的数据 防止OOM
                    txMap.remove(value.getTxId());
                } else {
                    //没有：将自己存入自己的缓存等待对方数据关联
                    orderMap.put(value.getTxId(), value);
                }
            }
            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                //1.先判断对方缓存中是否有能关联上的数据
                if (orderMap.containsKey(value.getTxId())) {
                    //有：关联成功，结果打印至控制台
                    out.collect("订单：" + orderMap.get(value.getTxId()).getOrderId() + "对账成功！！！！！");
                    //并且移除对方缓存中已关联上的数据 防止OOM
                    orderMap.remove(value.getTxId());
                } else {
                    //没有：将自己存入自己的缓存等待对方数据关联
                    txMap.put(value.getTxId(), value);
                }
            }
        }).print();

        env.execute();
    }
}
